package rpc.core.plugin.impl.proxy;

import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import rpc.bootstrap.springboot.Bootstrap;
import rpc.core.FlowReq;
import rpc.core.InitClose;
import rpc.core.InvokeInfo;
import rpc.core.RetryConfig;
import rpc.core.RpcApp;
import rpc.log.LogFormatManager;
import rpc.trace.Trace;
import rpc.utils.ExceptionUtils;
import rpc.utils.FileUtils;

public class AsyncFlowService implements InitClose {

  private static final Logger logger = LoggerFactory.getLogger(AsyncFlowService.class);
  private final Multimap<Long, InvokeInfo> delayedTaskMap = HashMultimap.create();
  private Bootstrap bootstrap;
  private RpcApp rpcApp;
  private ApplicationContext context;
  private ScheduledExecutorService executor;


  private String filePath;

  public Bootstrap getBootstrap() {
    return bootstrap;
  }

  public AsyncFlowService setBootstrap(Bootstrap bootstrap) {
    this.bootstrap = bootstrap;
    return this;
  }

  public RpcApp getRpcApp() {
    return rpcApp;
  }

  public AsyncFlowService setRpcApp(RpcApp rpcApp) {
    this.rpcApp = rpcApp;
    return this;
  }

  public ApplicationContext getContext() {
    return context;
  }

  public AsyncFlowService setContext(ApplicationContext context) {
    this.context = context;
    return this;
  }

  @Override
  public void init() {

    filePath = "/opt/data/" + rpcApp.getAppName() + "/task/async.cache";
    executor = new ScheduledThreadPoolExecutor(1,
        new ThreadFactoryBuilder().setNameFormat("pool_async_%d").build());
    executor.scheduleAtFixedRate(this::sendInvoke, 0, 1000, TimeUnit.MILLISECONDS);
    loadLocal();
  }

  @Override
  public void start() {

  }

  @Override
  public void close() {
    if (Objects.nonNull(executor)) {
      executor.shutdown();
    }
    writeLocal();
  }

  private void writeLocal() {
    FileUtils.write(filePath, delayedTaskMap);
  }

  private void sendInvoke() {

    long key = System.currentTimeMillis() / 1000;
    List<Long> keyStream = new ArrayList<>(this.delayedTaskMap.keySet()).stream().filter(t -> t <= key).collect(Collectors.toList());
    if (CollectionUtils.isNotEmpty(keyStream)) {
      Trace.getOrNew();
    }
    try {
      keyStream.stream().map(delayedTaskMap::get)
          .forEach(c -> c.forEach(invokeInfo -> {
            this.bootstrap.invokeFlow(invokeInfo.getFlowName(), invokeInfo.getReq()).thenAccept(flowRes -> {
              LogFormatManager.logger(invokeInfo.getFlowName(), invokeInfo.getReq(), flowRes);
              // 如果失败,并且 delayTimes.size()>0 加入下个 队列
              RetryConfig retryConfig = invokeInfo.getReq().getRetryConfig();
              if (flowRes.isFail()) {
                if (retryConfig.next()) {
                  long nextKey = retryConfig.nextKey();
//                  logger.info("nextKey:{}", nextKey);
                  this.delayedTaskMap.put(nextKey, invokeInfo);
                }
              }
            });
          }));
      keyStream.forEach(delayedTaskMap::removeAll);
    } catch (Exception e) {
      logger.error("exception:{} stackTrace:{}", e.getMessage(), ExceptionUtils.getTrace(e));
    }
    if (CollectionUtils.isNotEmpty(keyStream)) {
      Trace.clearTraceContext();
    }
  }

  private void loadLocal() {
    FileUtils.read(filePath, Multimap.class, true).ifPresent(delayedTaskMap::putAll);
  }

  public void addDelayedTask(String flowName, FlowReq req) {

    try {
      RetryConfig retryConfig = req.getRetryConfig();
      if (retryConfig.next()) {
        delayedTaskMap.put(retryConfig.nextKey(), new InvokeInfo(flowName, req));
      }
    } catch (Exception e) {
      logger.error("exception:{} stackTrace:{}", e.getMessage(), ExceptionUtils.getTrace(e));
    }
  }

}
